import asyncio
from typing import Optional
import aioredis
from aioredis import Redis, ConnectionPool
from tenacity import retry, stop_after_attempt, wait_fixed


class RedisManager:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis_url = redis_url
        self.pool: Optional[ConnectionPool] = None
        self.client: Optional[Redis] = None

    async def init_pool(self):
        """初始化连接池（新API）"""
        self.pool = ConnectionPool.from_url(
            self.redis_url,
            max_connections=50,
            socket_connect_timeout=5,
            socket_timeout=10,
            decode_responses=True,
            health_check_interval=30
            # max_idle_time=300
        )
        self.client = Redis(connection_pool=self.pool)

    async def close(self):
        """安全关闭连接池"""
        if self.pool:
            await self.pool.disconnect()

    @retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
    async def push_data(self, key: str, data: dict) -> str:
        """写入Redis Stream（带重试机制）"""
        async with self.client.pipeline(transaction=True) as pipe:
            msg_id = await pipe.xadd(
                key,
                data,
                maxlen=100000,
                approximate=True
            )
            await pipe.execute()
            return msg_id

    async def read_batch(self, key: str, batch_size=100) -> list:
        """分页读取Stream数据"""
        messages = await self.client.xread(
            streams={key: '0'},  # 从起始位置读取
            count=batch_size
        )
        return [
            (msg_id, data)
            for msg_id, data in messages[0][1]
        ]

    async def ack_data(self, key: str, msg_id: str):
        """消息确认机制"""
        await self.client.xack(key, msg_id)


async def main():
    redis_mgr = RedisManager(redis_url="redis://localhost:6379")

    try:
        await redis_mgr.init_pool()

        # 插入测试数据
        msg_id = await redis_mgr.push_data(
            "order_stream",
            {"user_id": "1002", "amount": "299.99", "status": "pending"}
        )
        print(f"插入成功，消息ID: {msg_id}")

        # 读取验证
        messages = await redis_mgr.read_batch("order_stream")
        print(f"最新数据: {messages}")

    except aioredis.RedisError as e:
        print(f"Redis操作异常: {str(e)}")
    finally:
        await redis_mgr.close()


if __name__ == "__main__":
    # async def connection_test():
    #     try:
    #         async with Redis.from_url("redis://localhost:6379") as r:
    #             latency = await r.ping()
    #             print(f"连接成功，延迟: {latency} ms")
    #     except Exception as e:
    #         print(f"连接失败: {type(e).__name__}: {str(e)}")
    # asyncio.run(connection_test())
    asyncio.run(main())